package com.rem.flink.flink3Transform;

import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 求出最活跃的用户 reduce规约操作
 *
 * @author Rem
 * @date 2022-09-28
 */

public class TransformReduceTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // env.setParallelism(1);

        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Bob", "./neek", 3000L),
                new Event("Jeck", "./ceel", 4000L),
                new Event("Linda", "./cart", 5000L),
                new Event("Bob", "./home", 6000L),
                new Event("Bob", "./cart", 2200L),
                new Event("Bob", "./index", 2300L),
                new Event("jesson", "./cart", 2400L),
                new Event("kid", "./cart", 500L));


        /**
         * 统计活跃次数最多的用户1
         */
        // stream.keyBy(Event::getUser).sum("count").print("sum");


        /**
         * 统计活跃次数最多的用户2
         */
        //reduceCount(stream);


        /**
         * 统计活跃时长最多的用户
         */
        //reduceTimestamp(stream);

        env.execute();

    }

    private static void reduceCount(DataStreamSource<Event> stream) {
        stream
                .keyBy(Event::getUser)
                .reduce((value1, value2) -> value2.setCount(value1.getCount() + value2.getCount())).print("reduce count");
    }

    private static void reduceTimestamp(DataStreamSource<Event> stream) {
        //第一步 按照user分组 累加timestamp
        SingleOutputStreamOperator<Tuple2<String, Long>> clickUser = stream.map(e -> new Tuple2<>(e.getUser(), e.getTimestamp()))
                .returns(Types.TUPLE(Types.STRING, Types.LONG))
                .keyBy(tuple2 -> tuple2.f0)
                .reduce((tupleA, tupleB) -> Tuple2.of(tupleA.f0, tupleA.f1 + tupleB.f1));


        //第二步 得出访问最大user
        clickUser.keyBy(a -> true)
                .reduce((tupleA, tupleB) -> tupleA.f1 > tupleB.f1 ? tupleA : tupleB)
                .print();

    }
}
